package vip.gnloypp.redis.mq.api.core.manager;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.bean.copier.CopyOptions;
import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.core.util.ArrayUtil;
import com.alibaba.fastjson.JSON;
import jakarta.annotation.PostConstruct;
import lombok.extern.log4j.Log4j2;
import org.springframework.data.domain.Range;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import vip.gnloypp.redis.mq.api.config.RedisReceiverConfig;
import vip.gnloypp.redis.mq.api.config.RedisReceiverQueueConfig;
import vip.gnloypp.redis.mq.api.core.base.LogManager;
import vip.gnloypp.redis.mq.api.data.RedisMqConstant;
import vip.gnloypp.redis.mq.api.data.RedisMqMessage;

import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * @author GuoNanLin
 * @since 2024/8/30 17:32
 */
@Log4j2
public class RedisReceiverManager {

    private final RedisReceiverConfig redisReceiverConfig;
    private final StringRedisTemplate stringRedisTemplate;
    private final LogManager logManager;

    public RedisReceiverManager(RedisReceiverConfig redisReceiverConfig) {
        this.redisReceiverConfig = redisReceiverConfig;
        this.stringRedisTemplate = new StringRedisTemplate(redisReceiverConfig.getRedisConnectionFactory());
        this.logManager = new LogManager(redisReceiverConfig.getLogPrefix());
    }

    @PostConstruct
    public void init() {
        redisReceiverConfig.getRedisReceiverQueueConfigs().forEach(config -> {
            // 必填校验
            logManager.notNull(config, "配置不能为空");
            logManager.notBlank(config.getQueue(), "queue不能为空");
            logManager.notNull(config.getReceiveHandler(), "receiveHandler不能为空，queue = {}", config.getQueue());
            // 默认值填充
            config.setGroup(Optional.ofNullable(config.getGroup()).orElse(RedisMqConstant.DEFAULT_GROUP));
            config.setConsumer(Optional.ofNullable(config.getConsumer()).orElse(RedisMqConstant.DEFAULT_CONSUMER));
            config.setBatchSize(Optional.ofNullable(config.getBatchSize()).orElse(RedisMqConstant.BATCH_SIZE));
            config.setRetryTimeout(Optional.ofNullable(config.getRetryTimeout()).orElse(RedisMqConstant.RETRY_TIMEOUT));
            config.setMinIdleTime(Optional.ofNullable(config.getMinIdleTime()).orElse(RedisMqConstant.MIN_IDLE_TIME));
            config.setRetryMaxCount(Optional.ofNullable(config.getRetryMaxCount()).orElse(RedisMqConstant.RETRY_MAX_COUNT));
        });
        for (RedisReceiverQueueConfig c : redisReceiverConfig.getRedisReceiverQueueConfigs()) {
            // redis key
            String key = redisReceiverConfig.getEnv() + ":mq:" + c.getQueue();
            // 创建消费者组
            this.createGroup(key, c.getGroup());
            // 消费消息
            this.receive(key, c);
            // 消费重试和队列长度修剪任务
            this.runTask(key, c);
        }
    }

    private void createGroup(String key, String group) {
        // 创建消费者组
        try {
            stringRedisTemplate.opsForStream().createGroup(key, group);
            logManager.info("消费者组，创建成功，key = {}，group = {}", key, group);
        } catch (Exception e) {
            logManager.info("消费者组，已存在，key = {}，group = {}", key, group);
        }
    }

    private <T extends RedisMqMessage> void receive(String key, RedisReceiverQueueConfig<T> config) {
        // 创建Stream消息监听容器配置
        StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>>
                options = StreamMessageListenerContainer.StreamMessageListenerContainerOptions.builder()
                // 一次性最多拉取多少条消息
                .batchSize(config.getBatchSize())
                // 执行消息轮询的执行器
                .executor(Executors.newFixedThreadPool(2))
                // 消息消费异常的handler
                .errorHandler(e -> log.error("StreamListener异常", e))
                // 超时时间，设置为0，表示不超时（超时后会抛出异常）
                .pollTimeout(Duration.ofMillis(20L))
                .build();

        // 创建Stream消息监听容器
        StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer =
                StreamMessageListenerContainer.create(this.redisReceiverConfig.getRedisConnectionFactory(), options);

        // 使用监听容器对象开始监听消费（使用的是手动确认方式），自动ack使用receiveAutoAck方法
        listenerContainer.receive(Consumer.from(config.getGroup(), config.getConsumer()),
                StreamOffset.create(key, ReadOffset.lastConsumed()),
                mapRecord -> {
                    T msg = BeanUtil.toBean(mapRecord.getValue(), config.getEventType(), CopyOptions.create());
                    msg.setKey(key);
                    msg.setMessageId(mapRecord.getId().getValue());
                    msg.setMessageDequeueCount(1L);
                    logManager.info("消费消息，msg = {}", JSON.toJSONString(msg));
                    Function<T, Boolean> function = config.getReceiveHandler()::receive;
                    if (Boolean.TRUE.equals(function.apply(msg))) {
                        stringRedisTemplate.opsForStream().acknowledge(config.getGroup(), mapRecord);
                    }
                });

        // 启动监听容器
        listenerContainer.start();
    }

    private <T extends RedisMqMessage> void runTask(String key, RedisReceiverQueueConfig<T> config) {
        Executors.newSingleThreadExecutor().execute(() -> {
            while (true) {
                try {
                    PendingMessages msgList = stringRedisTemplate.opsForStream().pending(key, config.getGroup()
                            , Range.unbounded(), config.getBatchSize());
                    if (!msgList.isEmpty()) {
                        // 消费重试
                        receivePendingMessages(msgList, key, config);
                    }
                } catch (Exception e) {
                    log.error("消费重试，失败，key = {}", key, e);
                } finally {
                    ThreadUtil.sleep(config.getRetryTimeout(), TimeUnit.SECONDS);
                }
            }
        });
    }

    /**
     * 消费重试
     *
     * @param msgList 重试列表
     * @param key     队列key
     * @param config  消费者配置
     */
    private <T extends RedisMqMessage> void receivePendingMessages(PendingMessages msgList, String key,
                                                                   RedisReceiverQueueConfig<T> config) {
        Function<T, Boolean> function = config.getReceiveHandler()::receive;
        logManager.info("消费重试，开始，size = {}", msgList.size());
        List<MapRecord<String, Object, Object>> records = stringRedisTemplate.opsForStream().claim(key,
                config.getGroup(), config.getConsumer(), Duration.ofSeconds(config.getMinIdleTime()),
                ArrayUtil.toArray(msgList.stream().map(PendingMessage::getId).toList(), RecordId.class));
        Map<String, MapRecord<String, Object, Object>> recordMap =
                records.stream().collect(Collectors.toMap(e -> e.getId().getValue(), Function.identity(),
                        (o1, o2) -> o1));
        for (PendingMessage msg : msgList) {
            MapRecord<String, Object, Object> mapRecord = recordMap.get(msg.getId().getValue());
            if (Objects.isNull(mapRecord)) {
                log.error("消费重试，消息不存在，key={}，recordId={}", key, msg.getId().getValue());
                continue;
            }
            T t = BeanUtil.toBean(mapRecord.getValue(), config.getEventType(), CopyOptions.create());
            t.setKey(key);
            t.setMessageId(msg.getId().getValue());
            long messageDequeueCount = msg.getTotalDeliveryCount() + 1L;
            t.setMessageDequeueCount(messageDequeueCount);
            if (messageDequeueCount <= config.getRetryMaxCount()) {
                try {
                    logManager.info("消费重试，msg = {}", JSON.toJSONString(t));
                    if (Boolean.TRUE.equals(function.apply(t))) {
                        stringRedisTemplate.opsForStream().acknowledge(key, config.getGroup(), msg.getId());
                    }
                } catch (Exception e) {
                    log.error("消费重试，失败，msg = {}", JSON.toJSONString(t), e);
                }
            } else {
                // 达到重试上限，打印日志，然后ack
                log.error("消费重试，次数达到上限，msg = {}", JSON.toJSONString(t));
                stringRedisTemplate.opsForStream().acknowledge(key, config.getGroup(), msg.getId());
            }
        }
    }

}
